package com.roy.wordcount;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Map;

public class CountBolt extends BaseRichBolt {
    OutputCollector collector;

    Map<String,Integer> map = new HashMap<>();
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    public void execute(Tuple tuple) {
        //按filed获取
        String word = tuple.getStringByField("word");
        //按角标获取
        int num = tuple.getInteger(1);
        if(map.containsKey(word)) {
            Integer count = map.get(word);
            map.put(word, count + num);
        }else{
            map.put(word,1);
        }
        System.out.println(map);
    }

    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
